package com.inf.utils;

import com.inf.vo.ResultPartnerVo;
import com.inf.vo.ParamsPartnerVo;
import com.inf.featureCompare.FeatureCompare;
import com.inf.vo.PersonBaseVo;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.util.Bytes;
import com.sun.jersey.core.util.Base64;
import com.inf.vo.BaseVo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;


/**
 * Created by zhangbn on 2018/10/23.
 */
public class HBaseUtil {

    static Logger log = LoggerFactory.getLogger(HBaseUtil.class);

    /**
     * 连接池对象
     */
    public static Connection connection;
    private static final String ZK_QUORUM = "hbase.zookeeper.quorum";
    private static final String ZK_CLIENT_PORT = "hbase.zookeeper.property.clientPort";
    private static final String HBASE_POS = "hsmaster";
    private static final String ZK_POS = "hsmaster,hsslave1,hsslave2";
    private final static String ZK_PORT_VALUE = "2181";

    /**
     * 静态构造
     */
    static {
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.rootdir", "hdfs://" + HBASE_POS + ":9000/hbase");
        configuration.set(ZK_QUORUM, ZK_POS);
        configuration.set(ZK_CLIENT_PORT, ZK_PORT_VALUE);
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 构造函数
     */
//    public HBaseUtil() {
//        Configuration configuration = HBaseConfiguration.create();
//        configuration.set("hbase.rootdir", "hdfs://" + HBASE_POS + ":9000/hbase");
//        configuration.set(ZK_QUORUM, ZK_POS);
//        configuration.set(ZK_CLIENT_PORT, ZK_PORT_VALUE);
//        try {
//            connection = ConnectionFactory.createConnection(configuration);
//        } catch (IOException e) {
//            e.printStackTrace();
//        }
//    }

    /**
     * 查询同行人（旧接口）
     *
     * @param tableName
     * @param start
     * @param end
     * @return
     * @throws IOException
     */
//    @Deprecated
//    public static List<PersonBaseVo> queryPartnerVo(String tableName, String start, String end) throws IOException {
//        List<PersonBaseVo> voList = new ArrayList<PersonBaseVo>();
//        Table table = connection.getTable(TableName.valueOf(tableName));
//        Scan scan = new Scan();
//        DateTimeFormatter format = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
//        DateTimeFormatter format_use = DateTimeFormatter.ofPattern("yyMMddHHmmss");
//        LocalDateTime startTime = LocalDateTime.parse(start, format);
//        LocalDateTime endTime = LocalDateTime.parse(end, format);
//        String startStr = startTime.format(format_use);
//        String endStr = endTime.format(format_use);
//        scan.setStartRow(Bytes.toBytes(startStr));
//        scan.setStopRow(Bytes.toBytes(endStr));
//        ResultScanner scanner = table.getScanner(scan);
//        for (Result result : scanner) {
//            String rowkey = Bytes.toString(result.getRow());
//            String uuid = Bytes.toString(result.getRow()).substring(12);
//            String pass_id = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("pass_id")));
//            String feature = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("feature")));
//            Double quality_score = Bytes.toDouble(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("quality_score")));
//            String face_img_url = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("face_img_url")));
//            String environ_img_url = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("environ_img_url")));
//            String pass_time = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("pass_time")));
//            String create_time = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("create_time")));
//            String camera_id = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("camera_id")));
//            Long camera_code = Bytes.toLong(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("camera_code")));
//            String device_id = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("device_id")));
//            String server_id = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("server_id")));
//            Integer age = Bytes.toInt(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("age")));
//            Integer gender = Bytes.toInt(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("gender")));
//            Integer attractive = Bytes.toInt(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("attractive")));
//            Integer eyeglass = Bytes.toInt(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("eyeglass")));
//            Integer sunglass = Bytes.toInt(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("sunglass")));
//            Integer smile = Bytes.toInt(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("smile")));
//            Integer mask = Bytes.toInt(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("mask")));
//            Integer race = Bytes.toInt(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("race")));
//            Integer eyeopen = Bytes.toInt(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("eyeopen")));
//            Integer mouthopen = Bytes.toInt(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("mouthopen")));
//            Integer beard = Bytes.toInt(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("beard")));
//
//            PersonBaseVo vo = new PersonBaseVo(rowkey, uuid, pass_id, pass_time, camera_id, camera_code, device_id, server_id, face_img_url, environ_img_url, quality_score, age, gender, attractive, eyeglass, sunglass, smile, mask, race, eyeopen, mouthopen, beard, feature, create_time);
//          //PersonBaseVo vo = new PersonBaseVo(rowkey, uuid, pass_id, pass_time, camera_id, face_img_url, environ_img_url, quality_score, feature);
//            voList.add(vo);
//        }
//        IOUtils.closeQuietly(table);
//        return voList;
//    }

    /**
     * 查询同行人（旧接口）
     *
     * @param tableName
     * @param paramVo
     * @return
     * @throws IOException
     */
    public List<ResultPartnerVo> queryPartnerVo(String tableName, ParamsPartnerVo paramVo) throws IOException {
        List<BaseVo> voList = new ArrayList<BaseVo>();
        Table table = connection.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        scan.setStartRow(Bytes.toBytes(paramVo.getBeginTime()));
        scan.setStopRow(Bytes.toBytes(paramVo.getEndTime()));
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            String row = Bytes.toString(result.getRow());
            String pass_id = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("pass_id")));
            //String feature = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("feature")));
            Double quality_score = Bytes.toDouble(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("quality_score")));
            String face_img_url = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("face_img_url")));
            String pass_time = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("pass_time")));
            String camera_id = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("camera_id")));
            BaseVo vo = new BaseVo(row, pass_id, pass_time, camera_id, quality_score, face_img_url);
            voList.add(vo);
        }
        IOUtils.closeQuietly(table);

        log.info("show: [" + paramVo.getBeginTime() + "-" + paramVo.getEndTime() + "]之间 voList.size = " + voList.size());

        byte[] var1 = Base64.decode(paramVo.getFeature());
        List<byte[]> listN = new ArrayList<byte[]>();
        listN = voList.stream().map(e -> Base64.decode(e.getFeature())).collect(Collectors.toList());
        double[] dArr = new FeatureCompare().featureCompByByte(var1, listN);

//        Double[] dArrD = ArrayUtils.toObject(dArr);
//        List dArrList = Arrays.asList(dArrD);
//        dArrList.forEach(System.out::println);
        Map<Integer, Double> rst = Tools.getMaxScoreAndIdx(dArr);
        List<BaseVo> voListRst = new ArrayList<BaseVo>();
        Iterator<Map.Entry<Integer, Double>> it = rst.entrySet().iterator();
        while (it.hasNext()) {
            Map.Entry<Integer, Double> next = it.next();
            if (next.getValue().compareTo(Double.valueOf(0.7)) > 0) {
                voListRst.add(voList.get(next.getKey()));
                log.info("show: " + "找到要搜索的图片的idx= " + next.getKey() + "    相似度=" + next.getValue());
                log.info("show: " + "找到要搜索的图片: " + voListRst.get(0));
            } else {
                log.info("show: " + "没有找到要搜索的图片,无同行人！");
            }
        }
        List<ResultPartnerVo> partnerRst = new ArrayList<>();
        if (voListRst.size() > 0) {
            String uuid = voListRst.get(0).getUuid();
            List<BaseVo> targetList = voList.stream().filter(x -> x.getUuid().equals(uuid)).collect(Collectors.toList());
            log.info("show: uuid= [" + uuid + "] 的目标图片size()= " + targetList.size());
            Map<ResultPartnerVo, Integer> partnerMap = new HashMap<>();
            for (BaseVo base : targetList) {
//                List<ResultPartnerVo> oriList = queryVoByRowKeyWithRange("bigdata:person-repo", base.getUuid(), 60, 60);
                List<ResultPartnerVo> oriList = new ArrayList<>();
                log.info("show: uuid=[" + base.getUuid() + "]的oriList.size()= " + oriList.size());
                Set<ResultPartnerVo> oriSet = new HashSet<>(oriList);
                log.info("show: uuid=[" + base.getUuid() + "]的oriSet.size()= " + oriSet.size());
                for (ResultPartnerVo k : oriSet) {
                    if (partnerMap.containsKey(k)) {
                        partnerMap.put(k, partnerMap.get(k) + 1);
                    } else {
                        partnerMap.put(k, 1);
                    }
                }
            }
            log.info("show: uuid= [" + uuid + "] 的可能同行人partnerMap.size()= " + partnerMap.size());
            Iterator<Map.Entry<ResultPartnerVo, Integer>> m_it = partnerMap.entrySet().iterator();
            while (m_it.hasNext()) {
                Map.Entry<ResultPartnerVo, Integer> entry = m_it.next();
                if (entry.getValue() >= 2) {
                    partnerRst.add(entry.getKey());
                }
            }
            log.info("show: 与uuid= [" + uuid + "] 同行" + paramVo.getCameraIds() + "次的人partnerRst.size()= " + partnerRst.size());
        }
        return partnerRst;
    }

    public static void main(String[] args) throws Exception {
        //hashMap key测试
//        BaseVo vo1 = new BaseVo("18072123354918072200004e4bce907028cc70042240f", "pass_id", "pass_time", "camera_id", 56.0, "face_img_url");
//        BaseVo vo2 = new BaseVo("18072123354918072200004e4bce907028cc70042240f", "pass_id", "pass_time", "camera_id", 56.0, "face_img_url");
//        Map<BaseVo, Integer> partnerMap = new HashMap<>();
//        partnerMap.put(vo1, 1);
//        System.out.println(partnerMap.size());
//        System.out.println(partnerMap);
//        System.out.println(partnerMap.containsKey(vo2));
//        Integer A = partnerMap.put(vo2, 2);
//        System.out.println(A);
//        System.out.println(vo1.hashCode() == vo2.hashCode());
//        System.out.println(partnerMap.size());
//        System.out.println(partnerMap);


        Table table = HBaseUtil.connection.getTable(TableName.valueOf("bigdata:person"));
        Scan scan = new Scan();
        DateTimeFormatter format = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        DateTimeFormatter format_use = DateTimeFormatter.ofPattern("yyMMddHHmmss");
        LocalDateTime startTime = LocalDateTime.parse("2018-07-22 00:00:00", format);
        LocalDateTime endTime = LocalDateTime.parse("2018-08-06 00:00:00", format);
        String startStr = startTime.format(format_use);
        String endStr = endTime.format(format_use);
        scan.setStartRow(Bytes.toBytes(startStr));
        scan.setStopRow(Bytes.toBytes(endStr));
        String uuid = "1807232023b44864a343e734c52e9950";
        Filter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator(".*" + uuid + "$"));
        scan.setFilter(filter);
        scan.setCaching(1000);
        scan.setBatch(10);
        ResultScanner scanner = table.getScanner(scan);
        int sum = 0;
        for (Result result : scanner) {
//            if(sum==50){
//                String rowkey = Bytes.toString(result.getRow());
//                System.out.println(rowkey);
//                String feature = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("feature")));
//                System.out.println(feature);
//            }
            sum++;
        }
        System.out.println(sum);
        IOUtils.closeQuietly(table);


    }


}
